package com.tanner.mapreducedemo.reducejoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * @author zhang.t.c
 * @date 2021/8/4
 */
public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {
    private String filename;
    private Text k = new Text();
    private TableBean v = new TableBean();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        InputSplit split = context.getInputSplit();
        FileSplit fileSplit = (FileSplit) split;
        filename = fileSplit.getPath().getName();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();

        if (filename.contains("order")) {
            String[] split = line.split("\t");
            String id = split[0];
            String pid = split[1];
            String amount = split[2];
            k.set(pid);
            v.setId(id);
            v.setPid(pid);
            v.setAmount(Integer.parseInt(amount));
            v.setPname("");
            v.setFlag("order");
        } else {
            String[] split = line.split("\t");
            String pid = split[0];
            String pname = split[1];
            k.set(pid);
            v.setId("");
            v.setPid(pid);
            v.setAmount(0);
            v.setPname(pname);
            v.setFlag("pd");
        }
        context.write(k,v);

    }

}
